package org.logsin37.tool.eurekamanager.infra.remote.impl;

import java.util.HashMap;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.apache.commons.lang3.StringUtils;
import org.logsin37.tool.eurekamanager.domain.entity.Eureka;
import org.logsin37.tool.eurekamanager.domain.entity.EurekaApplication;
import org.logsin37.tool.eurekamanager.domain.entity.EurekaInstance;
import org.logsin37.tool.eurekamanager.infra.constant.ErrorMessages;
import org.logsin37.tool.eurekamanager.infra.constant.EurekaEndpointConstants;
import org.logsin37.tool.eurekamanager.infra.constant.EurekaInstanceStatus;
import org.logsin37.tool.eurekamanager.infra.remote.EurekaRemoteClient;
import org.springframework.http.MediaType;
import org.springframework.util.Assert;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

/**
 * Eureka Remote Client Impl
 *
 * @author logsin37 2020/01/20 23:17
 */
@Slf4j
public class EurekaRemoteClientImpl implements EurekaRemoteClient {

    private final String eurekaCode;
    private WebClient webClient;
    private final ObjectMapper objectMapper;

    public EurekaRemoteClientImpl(Eureka eureka, ObjectMapper objectMapper) {
        Assert.notNull(eureka, String.format(ErrorMessages.NOT_NULL, "eureka info"));
        Assert.isTrue(StringUtils.isNotBlank(eureka.getCode()), String.format(ErrorMessages.NOT_NULL, "eureka code"));
        Assert.isTrue(StringUtils.isNotBlank(eureka.getAddress()), String.format(ErrorMessages.NOT_NULL, "eureka address"));
        this.eurekaCode = eureka.getCode();
        this.objectMapper = objectMapper;
        this.bind(eureka.getAddress());
    }

    @Override
    public String getEurekaCode(){
        return this.eurekaCode;
    }

    /**
     * bind aim eureka address
     *
     * @param aimEurekaBaseAddress aim eureka address
     */
    private void bind(String aimEurekaBaseAddress){
        this.webClient = WebClient
                .builder()
                .defaultHeader("Accept", MediaType.APPLICATION_JSON_VALUE)
                .baseUrl(aimEurekaBaseAddress)
                .build();
    }

    @Override
    public Flux<EurekaApplication> listAllApplications(){
        val jsonNode = this.webClient.get().uri(EurekaEndpointConstants.APPS).retrieve().bodyToMono(JsonNode.class);
        return jsonNode
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .map(root -> {
                    val applicationsNode = root.get("applications");
                    if(applicationsNode == null){
                        return this.objectMapper.createArrayNode();
                    }
                    val applicationListNode = applicationsNode.get("application");
                    if(applicationListNode == null || !applicationListNode.isArray()){
                        return this.objectMapper.createArrayNode();
                    }
                    return (ArrayNode) applicationListNode;
                })
                .publishOn(Schedulers.elastic())
                .flatMapMany(applicationListNode -> Flux.create(emitter -> {
                    for (JsonNode node : applicationListNode) {
                        try {
                            emitter.next(this.objectMapper.readValue(node.toString(), new TypeReference<EurekaApplication>() { }));
                        } catch (JsonProcessingException e) {
                            emitter.error(e);
                        }
                    }
                    emitter.complete();
                }));
    }

    @Override
    public Mono<Void> putStatus(EurekaInstance instance, EurekaInstanceStatus status) {
        Assert.notNull(instance, String.format(ErrorMessages.NOT_NULL, "instance"));
        Assert.notNull(instance.getApp(), String.format(ErrorMessages.NOT_NULL, "instance"));
        Assert.notNull(instance.getInstanceId(), String.format(ErrorMessages.NOT_NULL, "instance"));
        Assert.notNull(status, String.format(ErrorMessages.NOT_NULL, "instance"));
        val requestParam = new HashMap<String, Object>(4);
        requestParam.put("applicationName", instance.getApp());
        requestParam.put("instanceId", instance.getInstanceId());
        requestParam.put("status", status);
        return this.webClient.put().uri(EurekaEndpointConstants.PUT_STATUS, requestParam)
                .exchange()
                .subscribeOn(Schedulers.elastic())
                .publishOn(Schedulers.elastic())
                .then();
    }

}
